package com.xf.day07_graphx

import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
/**
 *  图的关联 与 聚合操作
 *     1.包含了 类型转换
 */
object OperatorGraphAgg {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("testGraphx2").setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("ERROR")

    // 创建一个顶点集的 RDD,  VertexId 一个 Long 类型, 顶点属性是一个 二元组
    val users : RDD[( VertexId, (String, Int))] = sc.parallelize(
      Array(
        ( 1L, ("菲菲", 22)),
        ( 2L, ("乔乔", 17)),
        ( 3L, ("王强", 19)),
        ( 4L, ("王刚", 21)),
        ( 5L, ("李倩", 20)),
        ( 6L, ("林峰", 25))
      ))

    // 创建一个边集的 RDD
    val relationships :RDD[ Edge[Int]] = sc.parallelize(
      Array( Edge(1L ,3L, 15) ,
        Edge(2L ,1L, 10) ,
        Edge(2L ,3L, 21) ,
        Edge(2L ,4L, 22) ,
        Edge(3L ,6L, 10) ,
        Edge(4L ,5L, 17) ,
        Edge(5L ,6L, 20)
      ))

    // 定义边中 用户缺失时的 默认 (缺失) 用户
    val defaultUser = ("某某", 18)

    // 使用 users 和 relationships 两个RDD 实例化 Graph类建立一个 Graph对象
    val userGraph = Graph( users, relationships, defaultUser)

    // 创建 顶点RDD
    val join = sc.parallelize(
      Array[(Long, (String, Int ))](
           (1L, ("女生",5)),
           (2L, ("女生",5))
      ))

    /**
     * 输出 原有的所有的 顶点
     *    (1,(菲菲,22))
     *    (2,(乔乔,17))
     *    (3,(王强,19))
     *    (4,(王刚,21))
     *    (5,(李倩,20))
     *    (6,(林峰,25))
     */
    userGraph.vertices.collect().foreach(println)
    println(" ===================> ")
    //
    /**
     * 更新 图顶点属性 并输出更新后的图的所有顶点
     *
     * joinVertices 函数参数 说明:
     *   VD : 表示的是 userGraph 数据中的 顶点属性   (1,(菲菲,22))
     *   U  : 表示的是 join 数据中的 顶点属性        (1L, ("女生",5))
     *
     * 具体的结果为:
     *    (1,(女生菲菲,27))
     *    (2,(女生乔乔,22))
     *    (3,(王强,19))
     *    (4,(王刚,21))
     *    (5,(李倩,20))
     *    (6,(林峰,25))
     */
    userGraph.joinVertices(join )( (VertexId, VD, U)  => ( U._1 + VD._1, VD._2 + U._2) ).vertices.collect().foreach( println)
    println(" ===================> ")

    // 创建 顶点RDD
    val join1 = sc.parallelize( Array[(Long, String)]( (1L, "管理员"), (2L, "管理员")  ))

    /**
     * 更新 图顶点属性并输出 更新后的 图的所有顶点
     *
     * 具体的结果如下:
     *    (1,(菲菲,22,Some(管理员)))
     *    (2,(乔乔,17,Some(管理员)))
     *    (3,(王强,19,None))
     *    (4,(王刚,21,None))
     *    (5,(李倩,20,None))
     *    (6,(林峰,25,None))
     */
    userGraph.outerJoinVertices( join1 )( ( VertexId, VD, U) => ( VD._1, VD._2 , U ) ).vertices.collect().foreach( println )

  }

}
